package com.ibm.cps.servlet;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.kafka.KafkaProducer;
import com.ibm.cps.kafka.KafkaTopicConstructor;
import com.ibm.cps.kafka.KafkaTopicConstructorForMultiTenants;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.MetadataTopologySort;
import com.ibm.cps.message.validation.ValidatorUtil;
import com.ibm.cps.model.*;
import com.ibm.cps.processors.ProcessorsGroup;
import com.ibm.cps.test.PublishMonitor;
import com.ibm.factories.PublishProcessorFactory;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * Created by root on 12/3/15.
 */
public class ProcessorsImplementationV2 implements ServletTemplate{
    private Integer kafkaPartition;
    private Logger logger;
    private KafkaProducer producer;
    private IPersistency persistency;
    private ProcessorMetadtaValidator processorValidator;
    private KafkaTopicConstructorForMultiTenants kafkaTopicConstructorForMultiTenants = new KafkaTopicConstructorForMultiTenants();
    public ProcessorsImplementationV2() throws CPSException {
        kafkaPartition = LocalConfig.KAFKA_PARTITION;
        logger = LoggerFactory.getLogger(this.getClass());
        producer = new KafkaProducer(LocalConfig.KAFKA_BROKERS);
        logger.info("Connect to kafka " + LocalConfig.KAFKA_BROKERS);
        persistency = MongoPersistency.getInstace();
        logger.info("Connect to persistency " + LocalConfig.MONGO_CONNECT_STRING + ", database is "
                + LocalConfig.PERSISTENCY_DATABASE);
        processorValidator = ProcessorMetadtaValidator.getInstance();
    }

    /**
     * validate processor/group metadata
     *
     * @param object
     * @param type
     * @param source
     * @param tenantid
     * @return
     * @throws CPSException
     */
    private String validateMetadata(ObjectNode object, String type, String source, String tenantid)
            throws CPSException {
        String processorid = AssignIdFactory.getProcessorid(type, object);
        boolean success = processorValidator.validate(tenantid, processorid, object, source);
        if (!success) {
            return null;
        }
        return constructRetMsg(source, processorid, type);
    }

    /**
     * construct return message according to different conditions
     *
     * @param source
     * @param processorid
     * @param type
     * @return
     */
    private String constructRetMsg(String source, String processorid, String type) {
        if (ProcessorsGroup.isGroupType(type)) {
            ObjectNode retNode = JSONUtility.newObjectNode();
            retNode.put("id", processorid);
            return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, retNode);
        }
        if (source.equals(MessageFields.GROUP_TYPE_SQL) || source.equals(MessageFields.NODERED)) {
            return null;
        } else {
            return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, processorid);
        }
    }

    @Override
    public String doPost(Map<String, String> key2Value, String input) throws CPSException {
        ArrayNode retArray = JSONUtility.newArrayNode();
        ArrayList<ObjectNode> sendMsg = new ArrayList<ObjectNode>();
        ArrayList<ObjectNode> groups = new ArrayList<ObjectNode>();
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
        String source = key2Value.get(MessageFields.SOURCE);
        if (source == null) {
            source = "rest";
        }
        try {
            DataSourceMonitor monitor = DataSourceMonitor.getInstance();
            PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
            MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
            input = adapter.adapt2Json(input, source, tenantid);
            if (input == null) {
                return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Null input metadata.");
            }

            logger.info("Adapter output: " + input);
            ArrayNode objects = JSONUtility.getArrayFromString(input);

            Collection<String> existedMetadatas = persistency.getProcessors(tenantid, null, null, null);
            ArrayNode arrayNode = JSONUtility.newArrayNode(existedMetadatas);
            if (arrayNode != null) {
                arrayNode.addAll(objects);
                MetadataTopologySort.getSortedMetadatas(arrayNode);
            }
            List<ObjectNode> metadatas = MetadataTopologySort.getSortedMetadatas(objects);
            for (ObjectNode object : metadatas) {
                String type = JSONUtility.GetString(MessageFields.TYPE, object);
                String result = validateMetadata(object, type, source, tenantid);
                if (!ProcessorsGroup.isGroupType(type)) {
                    sendMsg.add(object);
                } else {
                    groups.add(object);
                }
                if (result == null) {
                    continue;
                }
                ObjectNode node = JSONUtility.fromObjectString(result);
                retArray.add(node);
            }

            // if all metadata pass the validation, then send all of them to
            // persistency and kafka
            for (ObjectNode metadata : sendMsg) {
                //broadcast
                for(int i= 0;i<kafkaPartition;i++)
                {
                    producer.send(kafkaTopicConstructorForMultiTenants.getMetadataAddTopic(tenantid),
                            i,metadata.toString().getBytes());

                }

                persistency.putProcessor(metadata);
                logger.info("Put message into persistency:" + metadata);
            }

            for (ObjectNode groupMetadata : groups) {
                persistency.putGroupMetadata(groupMetadata);
                // persistency.putMetadata(tenantid, groupMetadata.toString());
            }

            return retArray.toString();
        } catch (CPSException e) {
            for (ObjectNode tmpObject : sendMsg) {
                processorValidator.deleteProcessor(tenantid, tmpObject.toString(), source);
            }
            for (ObjectNode tmpObject : groups) {
                processorValidator.deleteProcessor(tenantid, tmpObject.toString(), source);
            }
            throw e;
        }

    }

    @Override
    public String doGet(Map<String, String> key2Value) throws CPSException {
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
        String source = key2Value.get(MessageFields.SOURCE);
        if (source == null) {
            source = "rest";
        }
        DataSourceMonitor monitor = DataSourceMonitor.getInstance();
        PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
        MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
        Collection<String> metadatas = persistency.getProcessors(tenantid, key2Value.get(MessageFields.PROCESSORID),
                key2Value.get(MessageFields.TYPE), key2Value.get(MessageFields.GROUPID));
        // Collection<String> metadatas =
        // getMetadatas(key2Value.get(MessageFields.SOURCE), tenantid,
        // key2Value.get(MessageFields.TYPE),
        // key2Value.get(MessageFields.GROUPID),
        // key2Value.get(MessageFields.PROCESSORID));
        Collection<String> allMetadata = persistency.getProcessors(tenantid, null, null, null);
        // Collection<String> allMetadata =
        // persistency.getMetadataByUser(tenantid);
        Collection<String> result = adapter.getJsons(allMetadata, metadatas);
        return ServletUtil.constructCollectionJson(result);
    }

    @Override
    public String doDelete(Map<String, String> key2Value, String input) throws CPSException {
        throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "This kind of delete method is unsupported.");
    }

    private String deleteProcessors(String tenantid, Collection<String> deletedMetadata, String source)
            throws CPSException {
        String retMessage;
        ArrayNode deletedIds = JSONUtility.newArrayNode();
        if (deletedMetadata != null && deletedMetadata.size() >= 1) {
            for (String deleted : deletedMetadata) {
                ArrayNode tmp = processorValidator.deleteProcessor(tenantid, deleted, source);
                if (tmp != null && tmp.size() >= 1) {
                    for (int index = 0; index < tmp.size(); index++) {
                        deletedIds.add(tmp.get(index));
                    }
                    //broadcast
                    for(int i= 0;i<kafkaPartition;i++)
                    {
                        producer.send(kafkaTopicConstructorForMultiTenants.getMetadataDeleteTopic(tenantid),
                                i,deleted.getBytes());

                    }
                }
            }
            if (deletedIds.size() >= 1) {
                retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE,
                        deletedIds.toString() + " have been deleted.");
            } else {
                retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing has been deleted.");
            }
        } else {
            retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
        }
        return retMessage;
    }

    /**
     * delete sql group
     *
     * @param input
     * @throws CPSException
     */
    private void deleteGroupProcessors(String tenantid, String input, String source) throws CPSException {
        JsonNode arrayNode = JSONUtility.getArrayFromString(input);
        Set<String> groupids = new HashSet<String>();
        for (int i = 0; i < arrayNode.size(); i++) {
            ObjectNode element = (ObjectNode) arrayNode.get(i);
            String groupid = JSONUtility.GetString(MessageFields.GROUPID, element);
            if (groupid != null) {
                groupids.add(groupid);
            }
        }

        if (groupids.size() < 1) {
            throw new CPSException(ErrorCode.ERROR_FIELD_TYPE, "Can't get groupid from the adapter output " + input);
        }
        for (String groupid : groupids) {
            Collection<String> deletedProcessors = persistency.getProcessors(tenantid, null, null, groupid);
            Collection<String> groupMetadata = persistency.getGroupMetadata(tenantid, groupid);
            // Collection<String> deletedProcessors =
            // persistency.getGroupProcessors(tenantid, groupid);
            // String groupMetadata = persistency.getGroupById(tenantid,
            // groupid);
            if (groupMetadata == null) {
                logger.error("Group metadta of groupid " + groupid + " should be existed.");
            } else {
                deletedProcessors.addAll(groupMetadata);
            }
            deleteProcessors(tenantid, deletedProcessors, source);
        }
    }

    @Override
    public String doDelete(Map<String, String> key2Value) throws CPSException {
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
        Collection<String> deletedMetadata = ProcessorsGetFactory.getMetadatasAndGroups(key2Value.get(MessageFields.SOURCE),
                tenantid, key2Value.get(MessageFields.TYPE), key2Value.get(MessageFields.GROUPID),
                key2Value.get(MessageFields.PROCESSORID));
        String source = key2Value.get(MessageFields.SOURCE);
        if (deletedMetadata != null) {
            return deleteProcessors(tenantid, deletedMetadata, source);
        } else {
            return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
        }
    }

    @Override
    public String doPut(Map<String, String> key2Value, String input) throws CPSException {
        String tenantid = key2Value.get(MessageFields.TENANTID);
        ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);

        String source = key2Value.get("source");
        if (source == null) {
            source = "rest";
        }

        // if (!source.equals(MessageFields.SQL)) {
        // throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, source
        // + " can't do the put operation.");
        // }
        DataSourceMonitor monitor = DataSourceMonitor.getInstance();
        PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
        MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
        String adaptInput = adapter.adapt2Json(input, source, tenantid);

        if (adaptInput == null) {
            return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Null input metadata.");
        } else {
            logger.info("Adapter output: " + adaptInput);
            deleteGroupProcessors(tenantid, adaptInput, source);
            return doPost(key2Value, input);
        }
    }
}
